package com.withyou.euraka.provide.kfk;
import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import com.google.gson.Gson;
import com.withyou.euraka.provide.api.common.message.detail.OrderMessage;
/**
 * kfk 发送者
 * @author guicheng.huang
 * @date 2020年4月26日
 */
@Service
@Slf4j
public class KafkaSender {

	private Gson gson = new Gson();
	
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;	
	
	//异步发送
	public void sendOrderMessage(OrderMessage order) throws InterruptedException, ExecutionException{
		ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send("orderTopic", String.valueOf(order.getOrder().getId()), gson.toJson(order));
		send.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            //更新数据库
			@Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("async send message success partition [{}]", result.getRecordMetadata().partition());
                log.info("async send message success offest[{}]", result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("async send message fail [{}]", ex.getMessage());
            }
        });
	}
}
